package com.lvmama.pet.sync.service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.sql.DataSource;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.dbcp.BasicDataSource;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.lvmama.comm.bee.service.sync.SyncBaseExecuteService;
import com.lvmama.comm.bee.service.sync.SyncBaseNewService;
import com.lvmama.comm.bee.service.sync.SyncDataSourceService;
import com.lvmama.comm.bee.service.sync.SyncNewException;
import com.lvmama.comm.sync.po.SyncDataSource;
import com.lvmama.comm.sync.pojo.SyncTemplate;
import com.lvmama.comm.sync.pojo.SyncTemplateSql;
import com.lvmama.comm.sync.pojo.SyncTrigger;
import com.lvmama.comm.sync.utils.Constant;
import com.lvmama.comm.utils.MemcachedUtil;
import com.lvmama.pet.sync.service.util.SyncExecutor;

public class SyncBaseExecuteServiceImpl implements SyncBaseExecuteService {
	
	@Autowired
	private SyncBaseNewService syncBaseNewService;
	
	@Autowired
	private SyncDataSourceService syncDataSourceService;

	private static ConcurrentMap<String, DataSource> dataSourceMap = new ConcurrentHashMap<String, DataSource>();
	
	static final Log LOG = LogFactory.getLog(SyncBaseExecuteServiceImpl.class); 
	
	private static final int RESOURCE_EXECUTOR_THREAD_NUM = 100;
	
	private static final ExecutorService exec=Executors.newCachedThreadPool();

	
	public void init(){
		List<SyncDataSource> dataSources = syncDataSourceService.findAll();
		if(null!=dataSources&&dataSources.size()>0){
			for (SyncDataSource syncDataSource : dataSources) {
				addDataSource(syncDataSource);
			}
		}
		
	}
	
	public static void addDataSource(SyncDataSource syncDataSource){
		// 创建BasicDataSource对象
	    BasicDataSource ds = new BasicDataSource();
	    ds.setDriverClassName(syncDataSource.getDriveClass());
	    ds.setUrl(syncDataSource.getJdbcUrl());
	    ds.setUsername(syncDataSource.getUserName());
	    ds.setPassword(syncDataSource.getPassword());
	    ds.setInitialSize(10);
	    ds.setMaxActive(10);
	    ds.setMaxIdle(10);
	    ds.setMaxWait(10000);
	    ds.setTimeBetweenEvictionRunsMillis(30000);
	    ds.setMinEvictableIdleTimeMillis(30000);
	    ds.setRemoveAbandoned(true);
	    ds.setTestWhileIdle(true);
	    ds.setTestOnBorrow(false);
	    if(syncDataSource.getSourceType().equals("oracle")){
		    ds.setValidationQuery("SELECT 1 FROM SYS.DUAL");
	    }else{
		    ds.setValidationQuery("SELECT 1");

	    }
	    dataSourceMap.put(syncDataSource.getSourceCode(), ds);
	}
	
	public int testSyncTemplateSql(Long templateSqlId, SyncTrigger trigger) {
		SyncTemplateSql sql = syncBaseNewService.findSyncTemplateSqlById(templateSqlId);
		try {
			SyncDataContext.put(SyncDataContext.TRIGGER_DATA, trigger);
			return execute(sql, trigger.getPkValue(), true);
		} catch (Exception e) {
			throw new SyncNewException(e);
		} finally {
			SyncDataContext.remove();
		}
	}
	
	
	public void sync(SyncTrigger trigger) {
		try {
			SyncDataContext.put(SyncDataContext.TRIGGER_DATA, trigger);
			
			SyncTemplate uniqueKey = new SyncTemplate();
			uniqueKey.setDataSourceId(trigger.getDataSourceId());
			uniqueKey.setTableName(trigger.getTableName());
			uniqueKey.setUserName(trigger.getUserName());
			//添加缓存  
			String template_key = "TEMPLATE_"+trigger.getDataSourceId()+"_"+trigger.getTableName()+"_"+trigger.getUserName();
			SyncTemplate template = (SyncTemplate) MemcachedUtil.getInstance().get(template_key);
			if(template==null){
					template = syncBaseNewService.findSyncTemplateByUniqueKey(uniqueKey);
					MemcachedUtil.getInstance().set(template_key,300, template);
			}
			LOG.info("template_key:"+template_key+"----template:"+template);
			if (template == null) {
				throw new SyncNewException("TRIGGER ID[" + trigger.getTriggerId() + "]没有找到对应模板");
			}
			if ("false".equals(template.getStatus())) {
				throw new SyncNewException("TRIGGER ID[" + trigger.getTriggerId() + "]模板[" + template.getTemplateName() + "]已经禁用");
			}
			SyncTemplateSql temp = new SyncTemplateSql();
			temp.setTemplateId(template.getTemplateId());
			temp.setTriggerType(trigger.getTriggerType());
			String template_sql_key = "TEMPLATESQL_"+template.getTemplateId()+"_"+trigger.getTriggerType();
			List<SyncTemplateSql> targetSqls = (List<SyncTemplateSql>) MemcachedUtil.getInstance().get(template_sql_key);
			if(targetSqls== null){
				 targetSqls = syncBaseNewService.findTargetSyncTemplateSqlListByTemplateSql(temp);
				 MemcachedUtil.getInstance().set(template_sql_key,300,targetSqls);
			}
			LOG.info("template_sql_key:"+template_sql_key+"----targetSqls:"+targetSqls);
			if (targetSqls.size() == 0) {
				throw new SyncNewException("TRIGGER ID[" + trigger.getTriggerId() + "]模板[" + template.getTemplateName() + "]脚本未配置");
			}
			for (SyncTemplateSql targetSql : targetSqls) {
				try {
					execute(targetSql, trigger.getPkValue(), false);
				} catch (SyncNewException e) {
					throw new SyncNewException("TRIGGER ID[" + trigger.getTriggerId() + "]模板脚本 ID[" + targetSql.getTemplateSqlId() + "]\r\n" + e.getMessage());
				} catch (Exception e) {
					throw new SyncNewException("TRIGGER ID[" + trigger.getTriggerId() + "]模板脚本 ID[" + targetSql.getTemplateSqlId() + "]\r\n" + e.toString());
				}
			}
		} finally {
			SyncDataContext.remove();
		}
	}
	
	public void batchSync(SyncTrigger trigger) {
			 exec.submit(new batchExceuteSyncTrigger(trigger));
			 //exec.shutdown();
	}
	
	
	
	
	@SuppressWarnings("rawtypes")
	private int execute(SyncTemplateSql sql, String sourceId, boolean isTest) throws Exception {
		if (sql == null) {
			throw new SyncNewException("没有找到目标脚本");
		}
		if (StringUtils.trimToNull(sql.getSourceSqlId()) == null) {
			throw new SyncNewException("没有找到源脚本");
		}
		SyncTemplateSql sourceSql = syncBaseNewService.findSyncTemplateSqlById(Long.parseLong(sql.getSourceSqlId()));
		if (sourceSql == null) {
			throw new SyncNewException("没有找到源脚本");
		}
		DataSource ds = dataSourceMap.get(sourceSql.getDataSourceId());
		if (ds == null) {
			List<SyncDataSource> dataSources = syncDataSourceService.findBySourceCode(sql.getDataSourceId());
			addDataSource(dataSources.get(0));
			ds= this.dataSourceMap.get(sql.getDataSourceId());
			if(ds==null){
				throw new SyncNewException("没有找到源脚本数据源");
			}
		}
		try {
			int rows = 0;
			List list;
			if (Constant.SYNC_SCRIPT_TYPE.GROOVY.name().equals(sourceSql.getScriptType())) {
				list = SyncExecutor.getDefault().executeScriptQuery(sourceSql.getSqlContent(), sourceId, dataSourceMap);
			} else {
				if(isTest){
					list = SyncExecutor.getDefault().executeSQLQueryTest(sourceSql.getSqlContent(), sourceId, ds);
				}else{
					list = SyncExecutor.getDefault().executeSQLQuery(sourceSql.getSqlContent(), sourceId, ds);
				}
			}
			if (list == null) {
				return rows;
			}
			for (Object obj : list) {
				ds = dataSourceMap.get(sql.getDataSourceId());
				if (ds == null) {
					//如果ds为null 从数据库在加载一次 获取不到 抛出异常 解决多应用服务器同步
					List<SyncDataSource> dataSources = syncDataSourceService.findBySourceCode(sql.getDataSourceId());
					addDataSource(dataSources.get(0));
					ds= this.dataSourceMap.get(sql.getDataSourceId());
					if(ds==null){
						throw new SyncNewException("没有找到目标脚本数据源");
					}
				}
				if (Constant.SYNC_SCRIPT_TYPE.GROOVY.name().equals(sql.getScriptType())) {
					rows += SyncExecutor.getDefault().executeScriptUpdate(sql.getSqlContent(), obj, dataSourceMap, sourceId);
				} else {
					rows += SyncExecutor.getDefault().executeSQLUpdate(sql.getSqlContent(), obj, ds, sourceId);
				}
			}
			return rows;
		} catch (Exception e) {
			throw e;
		}
	}


	public void setsyncBaseNewService(SyncBaseNewService syncBaseNewService) {
		this.syncBaseNewService = syncBaseNewService;
	}

	public void setDataSourceMap(ConcurrentMap<String, DataSource> dataSourceMap) {
		this.dataSourceMap = dataSourceMap;
	}

	
	private class  batchExceuteSyncTrigger implements Runnable{

		private SyncTrigger trigger;
		
		
		public batchExceuteSyncTrigger(SyncTrigger syncTrigger) {
			this.trigger = syncTrigger;
		}


		@Override
		public void run() {
	        List<String> sucList = new ArrayList<String>();
	        Long  completeCount = 0l;
			try {	
				SyncDataContext.put(SyncDataContext.TRIGGER_DATA, trigger);
				SyncTemplate uniqueKey = new SyncTemplate();
				uniqueKey.setDataSourceId(trigger.getDataSourceId());
				uniqueKey.setTableName(trigger.getTableName());
				uniqueKey.setUserName(trigger.getUserName());
				SyncTemplate template = syncBaseNewService.findSyncTemplateByUniqueKey(uniqueKey);
				if (template == null) {
					throw new SyncNewException("TRIGGER ID[" + trigger.getTriggerId() + "]没有找到对应模板");
				}
				if ("false".equals(template.getStatus())) {
					throw new SyncNewException("TRIGGER ID[" + trigger.getTriggerId() + "]模板[" + template.getTemplateName() + "]已经禁用");
				}
				SyncTemplateSql temp = new SyncTemplateSql();
				temp.setTemplateId(template.getTemplateId());
				temp.setTriggerType(trigger.getTriggerType());
				List<SyncTemplateSql> targetSqls = syncBaseNewService.findTargetSyncTemplateSqlListByTemplateSql(temp);
				if (targetSqls.size() == 0) {
					throw new SyncNewException("TRIGGER ID[" + trigger.getTriggerId() + "]模板[" + template.getTemplateName() + "]脚本未配置");
				}
				Long value = 0l;
				Long templateSqlId = 0l;
				Long successNum = 0l;
				Long failNum =  0l;
				for(int i=0;i<trigger.getPkQuantity();i++){
					try {
							if(i==0){
								 value = Long.parseLong(trigger.getPkValue());
							}else{
								 value = value+1;
							}
							trigger.setPkValue(String.valueOf(value));
							for (SyncTemplateSql targetSql : targetSqls) {
								 templateSqlId = targetSql.getTemplateSqlId();
								 execute(targetSql, value.toString(), false);
							}
							completeCount = (value-Long.parseLong(trigger.getPkValue()))+1;
							successNum++;
							//更新syncTrigger completeCount
					} catch (Exception e) {
						SyncTrigger dest =  new SyncTrigger();
						try {
							PropertyUtils.copyProperties(dest, trigger);
							dest.setPkBegin(null);
							dest.setPkQuantity(null);
							dest.setSuccessNum(null);
							dest.setFailNum(null);
							dest.setRemark(e.getMessage());
							syncBaseNewService.saveSyncTrigger(dest);
							System.out.println(dest.toString());
						}catch (Exception e1) {
						
							LOG.error(e.getMessage(),e1);
						}
						failNum++;
						//批量执行失败 重新插入一条记录
						syncBaseNewService.saveSyncTrigger(trigger);
						LOG.error("TRIGGER ID[" + trigger.getTriggerId() + "]模板脚本 ID[" + templateSqlId + "]\r\n" + e.getMessage());
						continue;
						
					}
				}
				
				sucList.add(trigger.getTriggerId());
				
			} finally {
				if (sucList.size() > 0) {
	        		syncBaseNewService.updateTriggersStatus(sucList, Constant.SYNC_STATUS.SUCCESS.name());
	        	}if(completeCount==0l){
	        		syncBaseNewService.updateTriggersStatus(sucList, Constant.SYNC_STATUS.FAIL.name());
	        	}
				SyncDataContext.remove();
			}
			
		}


		public SyncTrigger getTrigger() {
			return trigger;
		}


		public void setTrigger(SyncTrigger trigger) {
			this.trigger = trigger;
		}
		
	}
	
	
public static void main(String[] args) {
	Long a=0l;
	for(int i=0;i<100;i++){
		a++;
	}
	System.out.println(a);
}
}
